package cn.itcast.flink.base;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.Random;

/**
 * Author itcast
 * Date 2021/8/1 9:22
 * 需求分析步骤：
 * 数据流中的数据
 * {"userID": "user_3", "eventTime": "2019-08-17 12:19:47", "eventType": "browse", "productID": 1}
 * {"userID": "user_2", "eventTime": "2019-08-17 12:19:48", "eventType": "click", "productID": 1}
 * mysql 数据库中的数据，三个字段
 * INSERT INTO `user_info` VALUES ('user_1', '张三', 10);
 * INSERT INTO `user_info` VALUES ('user_2', '李四', 20);
 * INSERT INTO `user_info` VALUES ('user_3', '王五', 30);
 * INSERT INTO `user_info` VALUES ('user_4', '赵六', 40);
 * 最终得到的数据集
 * (user_3,2019-08-17 12:19:47,browse,1,王五,33)
 * (user_2,2019-08-17 12:19:48,click,1,李四,20)
 */
public class BroadcastStateDemo {
    public static void main(String[] args) throws Exception {
        //1.创建流执行环境，设置并行度，开启chk，设置状态后端设置和chk属性，重启策略
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(1000);
        env.setStateBackend(new FsStateBackend("hdfs://node1:8020/flink-checkpoints"));
        /// 设置checkpoint参数
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
        );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        /// 如果setTolerableCheckpointFailureNumber为0 ，不容忍报错，如果报错当前程序结束
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(10);
        env.getCheckpointConfig().setCheckpointInterval(500);
        /// 固定重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 3000L));

        //2.自定义数据源-数据流（随机模拟用户点击的数据流和读取数据库中数据）
        //2-1.自定义实时数据流，设置随机数据并生成 Tuple4<String, String, String, Integer> 类型
        DataStreamSource<Tuple4<String, String, String, Integer>> userClickSource = env.addSource(new UserClickData());
        //2-2.从MySQL中读取用户的配置数据
        //+--------+----------+---------+
        //| userID | userName | userAge |
        //+--------+----------+---------+
        //| user_1 | 张三     |      10 |
        //| user_2 | 李四     |      20 |
        //| user_3 | 王五     |      30 |
        //| user_4 | 赵六     |      40 |
        //+--------+----------+---------+
        DataStreamSource<HashMap<String, Tuple2<String, Integer>>> userConfigSource = env.addSource(new UserConfigData());
        //3.transformation 转换
        //3-1.定义状态描述器-将配置流作为状态广播  HashMap<String, Tuple2<String, Integer>
        MapStateDescriptor<Void,HashMap<String,Tuple2<String,Integer>>> userConfigStateDesc = new MapStateDescriptor("userConfigState", Types.VOID, Types.MAP(Types.STRING, Types.TUPLE(Types.STRING, Types.INT)));
        //3-2.将用户配置数据源广播出去 configDS.broadcast(descriptor);
        BroadcastStream<HashMap<String, Tuple2<String, Integer>>> broadcastDataStream = userConfigSource.broadcast(userConfigStateDesc);
        //3-3.将用户数据流connect用户配置广播流
        SingleOutputStreamOperator<Tuple6<String, String, String, Integer, String, Integer>> resultDataStream = userClickSource.connect(broadcastDataStream)
                //3-4.处理(process)连接后的流-根据配置流补全事件流中的用户的信息,Tuple4和Tuple2合并
                .process(new BroadcastProcessFunction<Tuple4<String, String, String, Integer>, HashMap<String, Tuple2<String, Integer>>, Tuple6<String, String, String, Integer, String, Integer>>() {
                    //3-4-1.processElement 将广播变量中的用户配置信息和用户数据流合并成Tuple6<String, String, String, Integer, String, Integer>
                    @Override
                    public void processElement(Tuple4<String, String, String, Integer> value, ReadOnlyContext ctx, Collector<Tuple6<String, String, String, Integer, String, Integer>> out) throws Exception {
                        //获取 userID 来关联用户配置信息中的 用户名称和用户年龄
                        String userID = value.f0;
                        //读取广播变量 - 用户的配置信息
                        HashMap<String, Tuple2<String, Integer>> userConfigMap = ctx.getBroadcastState(userConfigStateDesc).get(null);
                        // 根据userID 来读取用户配置
                        if (userConfigMap != null) {
                            Tuple2<String, Integer> userNameAndAge = userConfigMap.get(userID);
                            if (userNameAndAge != null) {
                                String userName = userNameAndAge.f0;
                                int userAge = userNameAndAge.f1;
                                out.collect(Tuple6.of(userID, value.f1, value.f2, value.f3, userName, userAge));
                            }
                        }
                    }

                    //3-4-2.processBroadcastElement 将广播变量状态里的用户配置数据进行更新
                    @Override
                    public void processBroadcastElement(HashMap<String, Tuple2<String, Integer>> value, Context ctx, Collector<Tuple6<String, String, String, Integer, String, Integer>> out) throws Exception {
                        BroadcastState broadcastState = ctx.getBroadcastState(userConfigStateDesc);
                        //清空当前历史广播变量的值
                        broadcastState.clear();
                        //将当前的用户配置的数据重写写入到广播变量中
                        broadcastState.put(null, value);
                    }
                });
        //4.sinks 打印输出
        resultDataStream.print();
        //5.execute 执行环境
        env.execute();
    }

    /**
     * 用于生成用户的点击流数据
     */
    private static class UserClickData extends RichSourceFunction<Tuple4<String, String, String, Integer>> {
        boolean flag = true;
        Random rn = new Random();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String[] eventTypes = {"browse","click"};
        //run 方法中实现随机用户点击数据的生成
        @Override
        public void run(SourceContext<Tuple4<String, String, String, Integer>> ctx) throws Exception {
            while(flag){
                String userID = "user_" + rn.nextInt(4);
                String eventTime = sdf.format(System.currentTimeMillis());
                String eventType = eventTypes[rn.nextInt(2)];
                int productID = rn.nextInt(5);
                ctx.collect(Tuple4.of(userID,eventTime,eventType,productID));
                Thread.sleep(300);
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }

    /**
     * 创建一个读取mysql中用户配置的数据源
     */
    private static class UserConfigData extends RichSourceFunction<HashMap<String, Tuple2<String,Integer>>>{
        boolean flag = true;
        Connection conn = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        @Override
        public void open(Configuration parameters) throws Exception {
            //获取数据库连接
            conn = DriverManager.getConnection(
                    "jdbc:mysql://node3:3306/bigdata?useSSL=false",
                    "root",
                    "123456"
            );
            //编写SQL
            String sql = "SELECT userID,userName,userAge " +
                    "FROM user_info";
            //生成 prepareStatement
            ps = conn.prepareStatement(sql);
        }

        //需要每 5 分钟读取一次 mysql 数据
        @Override
        public void run(SourceContext<HashMap<String, Tuple2<String, Integer>>> ctx) throws Exception {
            while(flag){
                rs = ps.executeQuery();
                HashMap<String, Tuple2<String, Integer>> userConfig = new HashMap<>();
                while(rs.next()){
                    String userID = rs.getString("userID");
                    String userName = rs.getString("userName");
                    int userAge = rs.getInt("userAge");
                    userConfig.put(userID,Tuple2.of(userName,userAge));
                }
                ctx.collect(userConfig);
                Thread.sleep(300000);
            }
        }

        @Override
        public void cancel() {
            flag  = false;
        }

        @Override
        public void close() throws Exception {
            if(!rs.isClosed())rs.close();
            if(!ps.isClosed())ps.close();
            if(!conn.isClosed()) conn.close();
        }
    }
}
